/*
 * @Author: ygnnli 1585923072@qq.com
 * @Date: 2024-08-18 21:29:48
 * @LastEditors: ygnnli 1585923072@qq.com
 * @LastEditTime: 2024-08-21 21:16:21
 * @FilePath: /bitrpc/source/client/rpc_client.hpp
 * @Description: 这是默认设置,请设置`customMade`, 打开koroFileHeader查看配置 进行设置: https://github.com/OBKoro1/koro1FileHeader/wiki/%E9%85%8D%E7%BD%AE
 */
#include "requestor.hpp"
#include "rpc_caller.hpp"
#include "../common/dispatcher.hpp"
#include "rpc_registry.hpp"
#include "rpc_topic.hpp"

namespace bitrpc
{
    namespace client
    {
        // 服务注册客户端，用于注册服务
        class RegistryClient
        {
        public:
            using ptr = std::shared_ptr<RegistryClient>;
            // 构造函数传入注册中心的地址信息,用于连接注册中心
            RegistryClient(const std::string &ip, int port)
                : _requestor(std::make_shared<Requestor>()),
                  _provider(std::make_shared<client::Provider>(_requestor)),
                  _dispatcher(std::make_shared<Dispatcher>())
            {
                auto rsp_cb = std::bind(&client::Requestor::onResponse, _requestor.get(), std::placeholders::_1, std::placeholders::_2); // 响应的回调函数
                _dispatcher->registerhandler<BaseMessage>(MType::RSP_SERVICE, rsp_cb);                                                   // 注册rpc响应映射关系消息处理函数

                auto message_cb = std::bind(&Dispatcher::onMessage, _dispatcher.get(), std::placeholders::_1, std::placeholders::_2); // dispatcher.get()获取Dispatcher对象的指针
                _client = ClientFactory::create(ip, port);

                _client->setMessageCallback(message_cb);
                _client->connect();
            }
            // 注册服务，传入服务名和地址信息
            bool registryMethod(const std::string &method, const Address &host)
            {
                ELOG("注册服务:%s - %s", method.c_str(), host.first.c_str());
                return _provider->registryMethod(_client->connection(), method, host);
            }

        private:
            Requestor::ptr _requestor;       // 请求者
            client::Provider::ptr _provider; // 服务提供者
            Dispatcher::ptr _dispatcher;     // 调度器
            BaseClient::ptr _client;         // 客户端
        };

        // 服务发现客户端，用于发现服务
        class DiscoveryClient
        {
        public:
            using ptr = std::shared_ptr<DiscoveryClient>;
            // 构造函数传入注册中心的地址信息,用于连接注册中心
            DiscoveryClient(const std::string &ip, int port, const Discoverer::OfflineCallback &cb)
                : _requestor(std::make_shared<Requestor>()),
                  _discoverer(std::make_shared<client::Discoverer>(_requestor, cb)),
                  _dispatcher(std::make_shared<Dispatcher>())
            {
                auto rsp_cb = std::bind(&client::Requestor::onResponse, _requestor.get(), std::placeholders::_1, std::placeholders::_2); // 响应的回调函数
                _dispatcher->registerhandler<BaseMessage>(MType::RSP_SERVICE, rsp_cb);                                                   // 注册rpc响应映射关系消息处理函数

                auto req_cb = std::bind(&client::Discoverer::onServiceRequest, _discoverer.get(), std::placeholders::_1, std::placeholders::_2); // 响应的回调函数
                _dispatcher->registerhandler<ServiceRequest>(MType::REQ_SERVICE, req_cb);                                                        // 注册rpc响应映射关系消息处理函数

                auto message_cb = std::bind(&Dispatcher::onMessage, _dispatcher.get(), std::placeholders::_1, std::placeholders::_2); // dispatcher.get()获取Dispatcher对象的指针
                _client = ClientFactory::create(ip, port);

                _client->setMessageCallback(message_cb);
                _client->connect();
            }
            // 发现服务，向外提供服务发现接口
            bool serviceDiscovery(const std::string &method, Address &host)
            {
                DLOG("开始进行服务发现:%s!!", method.c_str());
                return _discoverer->serviceDiscovery(_client->connection(), method, host);
            }

        private:
            Requestor::ptr _requestor;           // 请求者
            client::Discoverer::ptr _discoverer; // 服务发现者
            Dispatcher::ptr _dispatcher;         // 调度器
            BaseClient::ptr _client;             // 客户端
        };

        class RpcClient
        {
        public:
            using ptr = std::shared_ptr<RpcClient>;
            // enableDiscovery--是否启用服务发现功能，也决定了传入的地址信息是注册中心的地址，还是服务提供者的地址
            RpcClient(bool enableDiscovery, const std::string &ip, int port)
                : _enableDiscovery(enableDiscovery),
                  _requestor(std::make_shared<Requestor>()),
                  _dispatcher(std::make_shared<Dispatcher>()),
                  _caller(std::make_shared<bitrpc::client::RpcCaller>(_requestor))
            {
                // 针对rpc请求后的响应进行的回调处理
                auto rsp_cb = std::bind(&client::Requestor::onResponse, _requestor.get(), std::placeholders::_1, std::placeholders::_2); // 响应的回调函数
                _dispatcher->registerhandler<BaseMessage>(MType::RSP_RPC, rsp_cb);                                                       // 注册rpc响应映射关系消息处理函数

                // 如果启用了服务发现，地址信息是注册中心的地址，是服务发现客户端需要连接的地址，则通过地址信息实例化discovery_client
                // 如果没有启用服务发现，则地址信息是服务提供者的地址，则直接实例化好rpc_client
                if (enableDiscovery)
                {
                    //服务下线的回调函数(delete)
                    auto offline_cb = std::bind(&RpcClient::delClient, this, std::placeholders::_1);
                    _discoverry_client = std::make_shared<DiscoveryClient>(ip, port, offline_cb);
                }
                else
                {
                    auto message_cb = std::bind(&Dispatcher::onMessage, _dispatcher.get(), std::placeholders::_1, std::placeholders::_2); // dispatcher.get()获取Dispatcher对象的指针
                    _rpc_client = ClientFactory::create(ip, port);

                    _rpc_client->setMessageCallback(message_cb);
                    _rpc_client->connect();
                }
            }

            // 同步调用
            bool call(const std::string &method, const Json::Value &params, Json::Value &result)
            {
                // DLOG("开始同步调用!");
                //  获取服务提供者的地址  1.服务发现 2.固定的服务提供者
                BaseClient::ptr client = getClient(method);
                if (client.get() == nullptr)
                {
                    ELOG("同步出错!");
                    return false;
                }
                // ELOG("同步没问题!");
                //  3.通过客户端连接,发送rpc请求
                return _caller->call(client->connection(), method, params, result);
            }

            // 异步调用
            bool call(const std::string &method, const Json::Value &params, RpcCaller::JsonAsyncResponse &result) // JsonAsyncResponse = std::future<Json::Value>;
            {
                // ELOG("开始异步调用!");
                //  获取服务提供者的地址  1.服务发现 2.固定的服务提供者
                BaseClient::ptr client = getClient(method);
                if (client.get() == nullptr)
                {
                    ELOG("异步出错!");
                    return false;
                }
                // ELOG("异步没问题!");
                //  3.通过客户端连接,发送rpc请求
                return _caller->call(client->connection(), method, params, result);
            }

            // 回调调用
            bool call(const std::string &method, const Json::Value &params, const RpcCaller::JsonResponseCallback &cb) // JsonResponseCallback = std::function<void(const Json::Value &)>;
            {
                // ELOG("开始回调调用!");
                //  获取服务提供者的地址  1.服务发现 2.固定的服务提供者
                BaseClient::ptr client = getClient(method);
                if (client.get() == nullptr)
                {
                    ELOG("回调出错!");
                    return false;
                }
                // ELOG("回调没问题!");
                //  3.通过客户端连接,发送rpc请求
                return _caller->call(client->connection(), method, params, cb);
            }

        private:
            // 创建客户端
            BaseClient::ptr newClient(const Address &host)
            {
                // ELOG("newClient进来了!");
                auto message_cb = std::bind(&Dispatcher::onMessage, _dispatcher.get(), std::placeholders::_1, std::placeholders::_2); // dispatcher.get()获取Dispatcher对象的指针
                auto client = ClientFactory::create(host.first, host.second);

                client->setMessageCallback(message_cb);
                client->connect();
                putClient(host, client); // 将客户端放入_rpc_clients
                return client;
            }
            // 获取客户端
            BaseClient::ptr getClient(const Address &host)
            {
                // ELOG("getClient(host)进来了!");
                std::unique_lock<std::mutex> lock(_mutex);
                auto it = _rpc_clients.find(host);
                if (it == _rpc_clients.end())
                {
                    return BaseClient::ptr();
                }
                // DLOG("获取到了客户端对象!");
                return it->second;
            }

            BaseClient::ptr getClient(const std::string method)
            {
                // ELOG("getClient(method)进来了!");
                BaseClient::ptr client;
                if (_enableDiscovery)
                {
                    // ELOG("注册中心启动了!")
                    //  1.通过服务发现,获取服务提供者的地址
                    Address host;
                    bool ret = _discoverry_client->serviceDiscovery(method, host);
                    if (ret == false)
                    {
                        ELOG("当前 %s 服务,没有找到服务提供者!", method.c_str());
                        return BaseClient::ptr();
                    }

                    // 2.查看服务提供者是否已有实例化客户端,有则直接使用,没有则实例化
                    client = getClient(host);
                    if (client.get() == nullptr)
                    {
                        ELOG("服务提供者是否已有实例化客户端!")
                        client = newClient(host);
                    }
                }
                else
                {
                    ELOG("是固定的服务提供者!")
                    client = _rpc_client; // 如果是固定的服务提供者,则直接使用
                }
                return client;
            }

            // 添加客户端
            void putClient(const Address &host, BaseClient::ptr client)
            {
                // ELOG("添加客户端进来了!")
                std::unique_lock<std::mutex> lock(_mutex);
                _rpc_clients.insert(std::make_pair(host, client));
            }
            // 删除客户端
            void delClient(const Address &host)
            {
                // ELOG("删除客户端进来了!")
                std::unique_lock<std::mutex> lock(_mutex);
                _rpc_clients.erase(host);
            }

        private:
            //第一个参数是Address,无法实现hash,所以需要自定义一个hash函数
            struct AddressHash
            {
                size_t operator()(const Address &host) const
                {
                    std::string addr = host.first + std::to_string(host.second);
                    return std::hash<std::string>{}(addr);
                }
            };
            Requestor::ptr _requestor;               // 请求者 需要用它来构造_discoverry_client
            bool _enableDiscovery;                   // 是否启用服务发现功能
            DiscoveryClient::ptr _discoverry_client; // 服务发现客户端
            RpcCaller::ptr _caller;                  // rpc调用者
            Dispatcher::ptr _dispatcher;             // 调度器
            BaseClient::ptr _rpc_client;             // 用于未启用服务发现
            std::mutex _mutex;                       // 互斥锁
            //<"127.0.0.1:8080", client1>
            std::unordered_map<Address, BaseClient::ptr, AddressHash> _rpc_clients; // 用于服务发现的客户端连接池
        };

        class TopicClient
        {
        public:
            TopicClient(const std::string &ip, int port)
                : _requestor(std::make_shared<Requestor>()),
                  _dispatcher(std::make_shared<Dispatcher>()),
                  _topic_manager(std::make_shared<TopicManger>(_requestor))
            {
                // 针对rpc请求后的响应进行的回调处理
                auto rsp_cb = std::bind(&client::Requestor::onResponse, _requestor.get(), std::placeholders::_1, std::placeholders::_2); // 响应的回调函数
                _dispatcher->registerhandler<BaseMessage>(MType::RSP_TOPIC, rsp_cb);                                                     // 注册rpc响应映射关系消息处理函数

                // 推送的回调
                auto msg_cb = std::bind(&TopicManger::onPublish, _topic_manager.get(), std::placeholders::_1, std::placeholders::_2);
                _dispatcher->registerhandler<TopicRequest>(MType::REQ_TOPIC, msg_cb);

                auto message_cb = std::bind(&Dispatcher::onMessage, _dispatcher.get(), std::placeholders::_1, std::placeholders::_2); // dispatcher.get()获取Dispatcher对象的指针
                _rpc_client = ClientFactory::create(ip, port);

                _rpc_client->setMessageCallback(message_cb);
                _rpc_client->connect();
            }

            // 创建主题
            bool create(const std::string &key)
            {
                return _topic_manager->create(_rpc_client->connection(), key);
            }

            // 删除主题
            bool remove(const std::string &key)
            {
                return _topic_manager->remove(_rpc_client->connection(), key);
            }

            // 订阅主题
            bool subscribe(const std::string &key, const TopicManger::SubCallback &cb)
            {
                return _topic_manager->subscribe(_rpc_client->connection(), key, cb);
            }

            // 取消订阅主题
            bool cancel(const std::string &key)
            {
                return _topic_manager->cancel(_rpc_client->connection(), key);
            }

            // 发布消息
            bool publish(const std::string &key, const std::string &msg)
            {
                return _topic_manager->publish(_rpc_client->connection(), key, msg);
            }

            void shutdown()
            {
                _rpc_client->shutdown(); // 关闭客户端
            }

        private:
            Requestor::ptr _requestor;       // 请求者
            TopicManger::ptr _topic_manager; // topic管理器
            Dispatcher::ptr _dispatcher;     // 调度器
            BaseClient::ptr _rpc_client;    // rpc客户端
            std::mutex _mutex;               // 互斥锁
        };
    }
}
